package com.yunhe.transfer.service.data.impl;

import com.yunhe.common.model.authority.Station;
import com.yunhe.transfer.client.AuthorityClient;
import com.yunhe.transfer.client.DeviceClient;
import com.yunhe.transfer.client.MeasHistoryClient;
import com.yunhe.transfer.client.MeasRealtimeClient;
import com.yunhe.common.constant.DataTransferConsts;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.constant.MeasurementConsts;
import com.yunhe.common.model.DomainList;
import com.yunhe.common.model.ResultObject;
import com.yunhe.common.model.authority.StationDTO;
import com.yunhe.common.model.data.Measurement;
import com.yunhe.common.model.data.MeasurementReqDTO;
import com.yunhe.common.model.device.Analog;
import com.yunhe.transfer.service.data.HeartbeatService;
import com.yunhe.common.util.DateTimeUtils;
import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
import com.yunhe.common.util.CollectionUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 心跳服务类
 * @author liuronglei
 */
@Service
public class HeartbeatServiceImpl implements HeartbeatService {
    @Autowired
    private MeasRealtimeClient measRealtimeClient;
    @Autowired
    private MeasHistoryClient measHistoryClient;
    @Autowired
    private DeviceClient deviceClient;
    @Autowired
    private AuthorityClient authorityClient;
//    @Autowired
//    @Qualifier("allStations")
//    private List<StationDTO> stations;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    // 记录心跳判断定时任务第一次启动时间
    private Long serverStartTime = null;

    @Override
    public void updateHeartbeatTimeMillis(Long stationCode) {
        updateHeartbeat(DataTransferConsts.getHeartbeatPid(stationCode), null, System.currentTimeMillis());
    }

    // 每5秒执行心跳判断
    @SchedulerLock(name = "HeartbeatTask", lockAtMostFor = "PT3S")
    @Scheduled(cron="0/5 * * * * ? ")
    public void heartbeatTask() {
        if (serverStartTime == null) {
            serverStartTime = System.currentTimeMillis();
        }
        Long propertyTimeout = 30000L;
        List<Station> stations = authorityClient.getStations(null).getResults();
        if (CollectionUtil.isNotEmpty(stations)) {
            for (Station station : stations) {
                Long stationCode = station.getId();
                // 如果电站code为空，或者不使用心跳逻辑，则跳过
                if (stationCode == null || station.getUseHeartbeat() == null || !station.getUseHeartbeat()) {
                    continue;
                }
                // 如果电站有超时时间就使用电站的超时时间
                if(station.getHeartbeatTimeout() != null){
                    propertyTimeout = station.getHeartbeatTimeout();
                }
                Long currentTimeMillis = System.currentTimeMillis();
                Long pointNumber = DataTransferConsts.getHeartbeatPid(stationCode);
                // 取上一次心跳状态
                ResultObject<Map<Long, Measurement>> preHeartbeat = measRealtimeClient.getValue(pointNumber).getBody();
                if (preHeartbeat.getResults() != null && preHeartbeat.getResults().size() > 0) {
                    Measurement preMeasurement = preHeartbeat.getResults().get(pointNumber);
                    if (preMeasurement != null) {
                        Double preValue = preMeasurement.getVal();
                        Long preTimeMillis = preMeasurement.getRtime();
                        if (preValue != null) {
                            // 如果服务器刚启动，并且上一次心跳状态是连接中，超时时间以启动时间为上次心跳时间
                            if (preTimeMillis < serverStartTime && preValue.equals(DataTransferConsts.Heartbeat.Connected.value())) {
                                preTimeMillis = serverStartTime;
                            }
                            // 心跳连接超时，并且上一次心跳状态不是断开，则变更状态，并执行连接断开操作
                            if (currentTimeMillis - preTimeMillis >= propertyTimeout
                                    && !preValue.equals(DataTransferConsts.Heartbeat.Disconnected.value())) {
                                updateHeartbeat(pointNumber, DataTransferConsts.Heartbeat.Disconnected, null);
                                disconnectedAfter(stationCode, pointNumber);
                            // 心跳连接未超时，并且上一次心跳状态不是连接中，则变更状态为已连接
                            } else if(currentTimeMillis - preTimeMillis < propertyTimeout
                                    && !preValue.equals(DataTransferConsts.Heartbeat.Connected.value())) {
                                updateHeartbeat(pointNumber, DataTransferConsts.Heartbeat.Connected, null);
                                connectedAfter(stationCode, pointNumber, false);
                            }
                        } else {
                            // 第一次连接时，直接变更状态为已连接
                            updateHeartbeat(pointNumber, DataTransferConsts.Heartbeat.Connected, null);
                            connectedAfter(stationCode, pointNumber, true);
                        }
                    }
                }
            }
        }
    }

    /**
     * 更新心跳值
     * @param pointNumber 点号
     * @param value 心跳值
     */
    private void updateHeartbeat(Long pointNumber, DataTransferConsts.Heartbeat value, Long rtime) {
        String dtime = DateTimeUtils.getDateString(rtime, null);
        measRealtimeClient.updateMeasureRealtime(pointNumber, rtime, dtime, value == null ? null : value.value());
    }

    /**
     * 获得需要清空值的点号列表
     * @param stationCode 电站代码
     * @return
     */
    private List<Analog> getNeedClearAnalog(Long stationCode) {
        return deviceClient.search(stationCode, false).getBody().getResults();
    }

    /**
     * 断开连接的处理
     * @param stationCode 电站代码
     * @param pointNumber 点号
     * @return
     */
    private void disconnectedAfter(Long stationCode, Long pointNumber) {
        // 可以发送Kafka消息，让其他程序来清空电站数据，也可以本定时任务直接清空，建议直接清空
        // 清空电站（或设备）对应的点号数据，并按当前最新值切出flag=1的历史量测
        List<Long> pidList = getStationPointNumber(stationCode);
        // 保存当前量测的切面值
        saveRealtime(pidList);
        // 将量测实时对象转为备份（用于重新连接时获取到断线前的最新值）
        realtimeMovetoBackup(pidList);
        // 发送心跳断开消息
        kafkaTemplate.send(KafkaConsts.Topic.StationDisconnected, String.valueOf(pointNumber));
    }

    /**
     * 重新连接的处理
     * @param stationCode 电站代码
     * @param pointNumber 点号
     * @param firstTime 是否第一次连接
     * @return
     */
    private void connectedAfter(Long stationCode, Long pointNumber, boolean firstTime) {
        // 第一次连接的时候，不需要进行备份
        if (!firstTime) {
            List<Long> pidList = getStationPointNumber(stationCode);
            // 保存备份值
            saveLastValue(pidList);
        }
        // 发送心跳连接消息
        kafkaTemplate.send(KafkaConsts.Topic.StationConnected, String.valueOf(pointNumber));
    }

    /**
     * 保存当前量测的切面值
     * @param pidList 点号列表
     * @return
     */
    private void saveRealtime(List<Long> pidList) {
        MeasurementReqDTO paramter = new MeasurementReqDTO();
        paramter.setPointNumber(pidList);
        Map<Long, Double> pidValueMap = measRealtimeClient.getMeasureRealtime(paramter).getBody().getResults();
        if (pidValueMap != null && pidValueMap.size() > 0) {
            Long rtime = System.currentTimeMillis();
            List<Measurement> measurementList = new ArrayList<>();
            for (Map.Entry<Long, Double> entry : pidValueMap.entrySet()) {
                if (entry.getValue() != null) {
                    // 保存不为空值的量测切面
                    Measurement measurement = new Measurement();
                    measurement.setPointNumber(entry.getKey());
                    measurement.setRtime(rtime);
                    measurement.setVal(entry.getValue());
                    measurement.setFlag(MeasurementConsts.FLAG.Disconnect.value());
                    measurementList.add(measurement);
                } else {
                    // 移除空值的点号
                    pidList.remove(entry.getKey());
                }
            }
            measHistoryClient.save(new DomainList<>(measurementList), MeasurementConsts.FREQUENCY.Original.value());
        }
    }

    /**
     * 将当前量测转为备份
     * @param pidList 点号列表
     * @return
     */
    private void realtimeMovetoBackup(List<Long> pidList) {
        MeasurementReqDTO map = new MeasurementReqDTO();
        map.setPointNumber(pidList);
        measRealtimeClient.movetoBackup(map);
    }

    /**
     * 保存并清空备份
     * @param pidList 点号列表
     * @return
     */
    private void saveLastValue(List<Long> pidList) {
        // 获得上次备份的数据，保存为重新上线量测
        MeasurementReqDTO paramter = new MeasurementReqDTO();
        paramter.setPointNumber(pidList);
        Map<Long, Double> pidValueMap = measRealtimeClient.getBackup(paramter).getBody().getResults();
        if (pidValueMap != null && pidValueMap.size() > 0) {
            Long rtime = System.currentTimeMillis();
            List<Measurement> measurementList = new ArrayList<>();
            for (Map.Entry<Long, Double> entry : pidValueMap.entrySet()) {
                if (entry.getValue() != null) {
                    // 保存不为空值的量测切面
                    Measurement measurement = new Measurement();
                    measurement.setPointNumber(entry.getKey());
                    measurement.setRtime(rtime);
                    measurement.setVal(entry.getValue());
                    measurement.setFlag(MeasurementConsts.FLAG.Connect.value());
                    measurementList.add(measurement);
                } else {
                    // 移除空值的点号
                    pidList.remove(entry.getKey());
                }
            }
            measHistoryClient.save(new DomainList<>(measurementList), MeasurementConsts.FREQUENCY.Original.value());
        }
        // 清空备份数据
        measRealtimeClient.clearBackup(paramter);
    }

    /**
     * 获得电站（设备）所有非指标类点号
     * @param stationCode 电站代码
     * @return
     */
    private List<Long> getStationPointNumber(Long stationCode) {
        // TODO 按照点号规则判断是电站心跳还是设备心跳
        // 获得电站（设备）所有非指标类点号
        List<Analog> analogList = getNeedClearAnalog(stationCode);
        List<Long> pidList = new ArrayList<>();
        if (analogList != null && analogList.size() > 0) {
            for (Analog analog : analogList) {
                pidList.add(analog.getPointNumber());
            }
        }
        return pidList;
    }
}
